EC2インスタンス起動時にInsufficientInstanceCapacityが発生したら別のインスタンスタイプで起動させてみる
こんにちは、muroです。AWS運用かんたん自動化ツール「opswitch」の開発・運用を担当しています。opswitchはboto3をがんがん使ってAWSを操作します。今回はboto3のモックライブラリ「moto」を利用して、EC2インスタンスの起動をテストしてみます。ただし、EC2インスタンスを起動するだけではなく、よくあるトラブルに対応した実装を試します。
前回記事はこちら。
シナリオ
EC2インスタンス起動時にInsufficientInstanceCapacityエラーが発生し、インスタンスを起動できないことがあります。
EC2 インスタンスの開始または起動時に発生する、InsufficientInstanceCapacity エラーのトラブルシューティング - AWS
数回リトライしても起動できなかった場合、別のインスタンスタイプに変更して起動させるようにしてみます。
環境
今回のテストに使用した環境は以下の通りです。
| 環境 | バージョン |
|---|---|
| Python | 3.9.13 |
| boto3 | 1.23.3 |
| moto | 3.1.9 |
| pytest | 7.1.2 |
| pytest-mock | 3.7.0 |
| retrying | 1.3.3 |
またテストプロジェクトの構成は以下の通りです。
. ├── main.py └── test_main.py
やってみる
テスト対象コード
import boto3
from retrying import retry
from botocore.exceptions import ClientError
import logging
logger = logging.getLogger()
def start_instances(instance_ids):
"""
EC2インスタンスを起動します。
"""
ec2 = boto3.client('ec2')
ec2.start_instances(
InstanceIds=instance_ids
)
def change_instance_type(instance_id, alt_instance_type):
"""
EC2インスタンスのインスタンスタイプを変更します。
"""
ec2 = boto3.client('ec2')
ec2.modify_instance_attribute(
InstanceId=instance_id,
Attribute='instanceType',
Value=alt_instance_type
)
@retry(wait_exponential_multiplier=1000, stop_max_attempt_number=5)
def start_instance(instance_id):
"""
EC2インスタンスの起動をリトライするラッパー関数です。
"""
logger.info('try start_instances...')
start_instances([instance_id])
def force_start_instance(instance_id, alt_instance_type):
"""
EC2インスタンスを起動します。
InsufficientInstanceCapacityエラー時はインスタンスタイプを変更して
再度EC2インスタンスの起動を試みます。
"""
try:
start_instance(instance_id)
except ClientError as ce:
logger.error(ce)
error_code = ce.response['Error']['Code']
if error_code == 'InsufficientInstanceCapacity':
change_instance_type(instance_id, alt_instance_type)
start_instance(instance_id)
retryingについて
関数に@retryデコレータを指定するだけでその関数をリトライしてくれる便利なライブラリです。今回はエクスポネンシャル・バックオフで時間をおいて最大5回リトライするように指定しました。retryingの詳しい使い方は以下の記事を参考にしてください。
エラーハンドリングについて
botocore.exceptions.ClientError をキャッチして、エラーコードがInsufficientInstanceCapacityエラーかどうかを判定しています。インスタンスタイプを変更して、さらにstart_instance()を呼び出しますが、それでも起動できなかった場合は例外が発生して終了します。boto3のエラーハンドリングについては以下のリファレンスも参考にしてください。
Error handling — Boto3 Docs 1.24.61 documentation
テストコード
今回もmotoを利用してEC2インスタンスをモックしています。さらに、pytest-mock を利用して、start_instances関数をモックしています。インスタンスタイプが代替指定のインスタンスタイプでない場合InsufficientInstanceCapacityエラーを発生させます。インスタンスタイプが代替指定のインスタンスタイプの場合にEC2インスタンスを起動して、結果をassertします。
import boto3
from moto import mock_ec2
import main
from botocore.exceptions import ClientError
import logging
logger = logging.getLogger()
@mock_ec2
def test_force_start_instances(mocker):
expected_number_of_instances = 1
initial_instance_type = 't2.medium'
input_alt_instance_type = 't3.medium'
input_instance_id = run_instances(
expected_number_of_instances,
initial_instance_type
)[0]
# 例外を発生させるようにMockを設定します。
def mock_start_instances(instance_ids):
instance = describe_instance(instance_ids[0])
instance_type = instance['InstanceType']
if instance_type == input_alt_instance_type:
ec2 = boto3.client('ec2')
ec2.start_instances(
InstanceIds=instance_ids
)
else:
raise ClientError(
{'Error': {'Code': 'InsufficientInstanceCapacity'}},
'StartInstances'
)
mocker.patch('main.start_instances', side_effect=mock_start_instances)
# テストを実行します。
main.force_start_instance(input_instance_id, input_alt_instance_type)
# インスタンスが起動されたかどうかテストします。
instance = describe_instance(input_instance_id)
assert(instance['InstanceType'] == input_alt_instance_type)
assert(instance['State']['Name'] == 'running')
def run_instances(expected_number_of_instances, instance_type):
# 期待するインスタンス数の分だけインスタンスを起動します。
ec2 = boto3.client('ec2')
images = ec2.describe_images()['Images']
response = ec2.run_instances(
ImageId=images[0]['ImageId'],
MinCount=expected_number_of_instances,
MaxCount=expected_number_of_instances,
InstanceType=instance_type
)
# インスタンスIDのリストを取得します。
instance_ids = [x['InstanceId'] for x in response['Instances']]
# インスタンスを停止します。
ec2.stop_instances(
InstanceIds=instance_ids
)
# インスタンスIDのリストを返します。
return instance_ids
def describe_instance(instance_id):
ec2 = boto3.client('ec2')
response = ec2.describe_instances(InstanceIds=[instance_id])
return response['Reservations'][0]['Instances'][0]
テスト実行
それではテストを実行してみます。今回のテスト対象コードはloggerを仕込んでいるので、その出力を表示させながら実行します。
python -m pytest --log-cli-level=INFO -s \ --log-cli-format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s" ========================================== test session starts ========================================== platform linux -- Python 3.9.13, pytest-7.1.2, pluggy-1.0.0 rootdir: /home/ユーザー/プロジェクト plugins: cov-3.0.0, mock-3.7.0 collected 1 item test_main.py::test_force_start_instances --------------------------------------------- live log call --------------------------------------------- 17:57:01.855 INFO Found credentials in environment variables. 17:57:02.041 INFO try start_instances... 17:57:04.072 INFO try start_instances... 17:57:08.139 INFO try start_instances... 17:57:16.193 INFO try start_instances... 17:57:32.242 INFO try start_instances... 17:57:32.299 ERROR An error occurred (InsufficientInstanceCapacity) when calling the StartInstances operation: Unknown 17:57:32.308 INFO try start_instances... PASSED ========================================== 1 passed in 31.20s ===========================================
2秒・4秒・8秒・16秒とリトライ間隔をあけてリトライされ、エラーの後テストが成功しました。
まとめ
retryingライブラリを利用することでエクスポネンシャル・バックオフで待機時間を延ばしながら関数をリトライさせることができました。また、motoでEC2インスタンスをモックしつつpytest-mockを利用して複雑なエラー条件をテストすることができました。なかなか発生させにくいエラーとそのハンドリングであっても、ライブラリを利用することで、ある程度再現させられるのはとても便利だと思いました。以上、ご参考になりましたら幸いです。
![[アップデート] EBSスナップショットとAMIのゴミ箱保持ルールの適用対象をタグで制御できるようになりました](https://images.ctfassets.net/ct0aopd36mqt/wp-thumbnail-bac3d29aa65f45576f73094798087ee5/039ad6f8a7d8f18da47986d21c447f48/amazon-ec2)
